package com.buddy.zookeeper.master;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Random;

public class MasterWatcher implements Watcher {

    private static final Logger LOG = LoggerFactory.getLogger(MasterAsync.class);


    ZooKeeper zk;
    String hostPort;
    String serverId = Integer.toString(new Random().nextInt());
    MasterStates state;

    AsyncCallback.StringCallback masterCreateCallback = new AsyncCallback.StringCallback() {
        public void processResult(int rc, String path, Object ctx, String name) {
            switch (KeeperException.Code.get(rc)) {
                case CONNECTIONLOSS:
                    checkMaster();
                    return;
                case OK:
                    state = MasterStates.ELECTED;
                    takeLeadership();
                    break;
                case NODEEXISTS:
                    state = MasterStates.NOTELECTED;
                    masterExists();
                default:
                    state = MasterStates.NOTELECTED;
                    LOG.error("Something went wrong when running for master.",
                            KeeperException.create(KeeperException.Code.get(rc), path));
            }
        }
    };

    AsyncCallback.DataCallback masterCheckCallBack = new AsyncCallback.DataCallback() {
        public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
        }
    };

    AsyncCallback.StatCallback masterExistsCallback = new AsyncCallback.StatCallback() {
        public void processResult(int rc, String path, Object ctx, Stat stat) {
            switch (KeeperException.Code.get(rc)) {
                case CONNECTIONLOSS:
                    masterExists();
                    return;
                case OK:
                    if (stat == null) {
                        state = MasterStates.RUNNING;
                        runForMaster();
                    }
                    break;
                default:
                    checkMaster();
                    break;
            }
        }
    };

    Watcher masterExistsWatcher = new Watcher() {
        public void process(WatchedEvent event) {
            if (event.getType() == Event.EventType.NodeDeleted) {
                runForMaster();
            }
        }
    };

    MasterWatcher(String hostPort) {
        this.hostPort = hostPort;
    }


    void runForMaster() {
        System.out.println("----------> create /master");
        zk.create("/master",
                serverId.getBytes(),
                ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.EPHEMERAL,
                masterCreateCallback,
                null);
    }

    void checkMaster() {
        zk.getData("/master", false, masterCheckCallBack, null);
    }

    void masterExists() {
        zk.exists("/master", masterExistsWatcher, masterExistsCallback, null);
    }

    void takeLeadership() {
    }


    void startZK() throws IOException {
        zk = new ZooKeeper(hostPort, 15000, this);
    }

    void stopZK() throws InterruptedException {
        zk.close();
    }

    public void process(WatchedEvent event) {

    }


    public static void main(String[] args) throws IOException, InterruptedException {
        MasterWatcher masterWatcher = new MasterWatcher("127.0.0.1:2181");
        masterWatcher.startZK();
        masterWatcher.runForMaster();
        Thread.sleep(60000);
        masterWatcher.stopZK();
    }
}
